package com.wsjj.yjh;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class OperatorListStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

//        env.fromSource(KafkaSource.builder().build(), WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(3)),"kafkasource");
        DataStreamSource<String> source = env.socketTextStream("1.94.41.70", 7777);
        SingleOutputStreamOperator<Long> map = source.map(new MyCountMapFunction());
        map.print();
        env.execute();

    }


    private static class MyCountMapFunction implements MapFunction<String,Long> , CheckpointedFunction {
        Long count = 0L;
        ListState<Long> state;

        @Override
        public Long map(String value) throws Exception {

            return count++;
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("snapshotState....");
            state.clear();
            state.add(count);

        }

        /**
         * 初始化方法
         * @param context
         * @throws Exception
         */
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState......");
            state = context.getOperatorStateStore().getListState(new ListStateDescriptor<Long>("name", Types.LONG));

            if (context.isRestored()) {
                for (Long c : state.get()) {
                    count += c;
                }
            }

        }
    }
}
